package cn.wangjie.spark.iot

import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery}
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 对物联网设备状态信号数据，实时统计分析:
 * 1）、信号强度大于30的设备
 * 2）、各种设备类型的数量
 * 3）、各种设备类型的平均信号强度
 */
object IotStreamingOnlineSQL {
	
	def main(args: Array[String]): Unit = {
		// 1. 创建SparkSession实例对象
		val spark: SparkSession = SparkSession
			.builder()
			.appName(this.getClass.getSimpleName.stripSuffix("$"))
			.master("local[2]")
			// 设置SparkSQL Shuffle时分区数目
			.config("spark.sql.shuffle.partitions", "2")
			.getOrCreate()
		// 导入隐式转换和函数库
		import org.apache.spark.sql.functions._
		import spark.implicits._   // 此函数库中所有函数针对DataFrame操作
		
		// TODO: 1. 从Kafka消费数据
		val kafkaStreamDF: DataFrame = spark
			.readStream
			.format("kafka")
			.option("kafka.bootstrap.servers", "node1.itcast.cn:9092")
			.option("subscribe", "iotTopic")
			.load()
		
		// TODO： 按照业务需求，实时分析数据
		val etlStreamDF: DataFrame = kafkaStreamDF
			// 将value转换为String字符串类型
			.selectExpr("CAST(value AS STRING)")
			// 将DataFrame转换为Dataset
			.as[String]
			// 数据样本：{"device":"device_10","deviceType":"db","signal":37.0,"time":1597571144413}
			// TODO: 调用函数，从json格式列中获取字段的值
    		.select(
			    get_json_object($"value", "$.device").as("device_id"), //
			    get_json_object($"value", "$.deviceType").as("device_type"), //
			    get_json_object($"value", "$.signal").as("signal"), //
			    get_json_object($"value", "$.time").as("time") //
		    )
			
		// TODO: 使用SQL对StreamingDataFrame进行分析
		// 第一步、注册DataFrame为临时视图
		etlStreamDF.createOrReplaceTempView("tmp_view_iots")
		// 第二步、编写SQL并执行
		val resultStreamDF: DataFrame = spark.sql(
			"""
			  |SELECT
			  |  device_type,
			  |  COUNT(device_type) AS device_count,
			  |  ROUND(AVG(signal), 2) AS device_signal
			  |FROM
			  |  tmp_view_iots
			  |WHERE
			  |  signal > 30
			  |GROUP BY
			  |   device_type
			  |""".stripMargin)
		
		// 输出数据和启动启用
		val query: StreamingQuery = resultStreamDF
			.writeStream
			// 使用Complete完全模式输出，将ResultTable中所有数据进行输出
			.outputMode(OutputMode.Complete())
			.format("console")
			.option("numRows", "20")
			.option("truncate", "false")
			// 针对流式应用，需要启动执行
			.start()
		query.awaitTermination() // 当流式应用运行以后，正常情况一直运行，除非人为或程序异常终止，等待结束
		query.stop()
	}
	
}
